package adapter

import (
	"context"
	"fmt"
	"ginfwk/infrastructure/component/config"
	"ginfwk/infrastructure/component/logger"
	"ginfwk/infrastructure/component/queue"
	amqp "github.com/rabbitmq/amqp091-go"
	"go.uber.org/zap"
	"sync"
	"time"
)

type AmqpAdapter struct {
	Ctx    context.Context
	Prefix string // 队列名前缀
	// 单独
	Logger *zap.SugaredLogger

	// 共用
	Cfg  *config.Queue
	Pool *sync.Pool
}

type AmqpConn struct {
	conn    *amqp.Connection
	channel *amqp.Channel
}

var _ queue.Queue = (*AmqpAdapter)(nil)

var DefaultAmqpAdapter *AmqpAdapter

// 全局队列初始化
func NewAmqpAdapter(cfg *config.Queue) queue.Queue {
	log := logger.Default()
	amqp.SetLogger(&AmqpLogger{Logger: log})

	amqpAdapter := &AmqpAdapter{
		Logger: log,
	}

	// 赋值配置,请求中再去获取连接
	amqpAdapter.Cfg = cfg

	amqpAdapter.initPool()

	// 全局队列
	DefaultAmqpAdapter = amqpAdapter
	return amqpAdapter
}

// CtxQueue 创建请求级的队列
func (q AmqpAdapter) CtxQueue(ctx context.Context, ctxLogger *zap.SugaredLogger) queue.Queue {
	ctxQueue := q
	ctxQueue.Ctx = ctx
	ctxQueue.SetLogger(ctxLogger)
	return &ctxQueue
}

// getConnection
// return amqp.Connection
func (q *AmqpAdapter) getConnection() any {
	cfg := q.Cfg.Amqp
	url := fmt.Sprintf("amqp://%s:%s@%s/", cfg.Username, cfg.Password, cfg.Addr)

	local := "zh-CN"
	if len(cfg.Locale) > 0 {
		local = cfg.Locale
	}

	heartbeat := 5
	if cfg.Heartbeat > 0 {
		heartbeat = cfg.Heartbeat
	}

	conn, err := amqp.DialConfig(url, amqp.Config{
		Vhost:     cfg.Vhost,
		Heartbeat: time.Duration(heartbeat) * time.Second,
		Locale:    local,
	})
	if err != nil {
		q.Logger.Errorf("can`t open connection for amqp, err: %s\n", err)
	}

	channel, err := conn.Channel()
	if err != nil {
		q.Logger.Errorf("can`t get channel from amqp connection, err: %s\n", err)
	}
	return &AmqpConn{
		conn:    conn,
		channel: channel,
	}
}

func (q *AmqpAdapter) getConnFromPool() *AmqpConn {
	return q.Pool.Get().(*AmqpConn)
}

func (q *AmqpAdapter) PutConnToPool(conn *AmqpConn) {
	q.Pool.Put(conn)
}

func (q *AmqpAdapter) initPool() {
	q.Pool = &sync.Pool{
		New: func() any {
			return q.getConnection()
		},
	}
}

func (q *AmqpAdapter) Send(queueName string, route string, params interface{}, options ...*queue.Options) error {
	option := &queue.Options{}
	if len(options) > 0 {
		option = options[0]
	}
	message := NewMessage(q.Prefix, queueName, route, params, option)

	conn := q.getConnFromPool()
	defer q.PutConnToPool(conn)

	cli := NewClient(conn)
	cli.SetLogger(q.Logger)

	// 创建队列并初始化
	if err := cli.initQueueAndMessage(message); err != nil {
		q.Logger.Errorf("failed to init queue or message, err:%s\n", err)
		return err
	}

	// 发布消息
	if err := cli.publishMessage(message); err != nil {
		q.Logger.Errorf("failed to publish message, err:%s\n", err)
		return err
	}
	return nil
}

// SetLogger 注入请求级logger
func (q *AmqpAdapter) SetLogger(log *zap.SugaredLogger) {
	q.Logger = log
	amqp.SetLogger(&AmqpLogger{Logger: log})
}

// SendDelay 发送延时消息
func (q *AmqpAdapter) SendDelay(queueName string, route string, param interface{}, delay time.Duration, options ...*queue.Options) error {
	if len(options) > 0 {
		option := options[0]
		option.DelayTime = delay
		options[0] = option
	} else {
		options = []*queue.Options{
			{DelayTime: delay},
		}
	}

	return q.Send(queueName, route, param, options...)
}
